package org.eliza.base.coroutinesx.channel

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*


/*通道(Channel)用于在多个协程之间传输数据。
Channel是和BlockingQueue非常相似的概念。不同的是写入和读取数据用的是异步的 send 和 receive

在通道没有数据时，调用recieve会导致协程挂起；在通道缓冲满时，调用send会导致协程挂起。
通道遵循FIFO原则，先发出的消息会先被获取。
*/





private fun channelNoBuffer() = runBlocking {
    val channel = Channel<Int>() // 无缓冲区 同步发送接收
    launch {// 子协程send
        // this might be heavy CPU-consuming computation or async logic,
        // we'll just send five squares
        for (x in 1..5) {
            println("send start $x")
            channel.send(x)
            println("send block end $x \n------------")
        }
        channel.close()
    }
    // here we print five received integers:
    // 主接收
    for (y in channel) {
        delay(500L)
        println("send receive ==> ${y}")
    }
    println("Done!")
}

private fun channelBuffer(time: Int) = runBlocking {
    val channel = Channel<Int>(time, onBufferOverflow = BufferOverflow.SUSPEND)
    launch {
        for (x in 1..10) {
            println("send start $x")
            channel.send(x)
            println("send block end $x \n------------")
        }
        channel.close()
    }
    for (y in channel) {
        println("send receive ==> ${y}")
    }
    println("Done!")
}
//===========


private fun channelProduceSquares(time: Int) = runBlocking {
    val squares = produceNumbers()
    //consumeEach 对每个接收到的元素执行给定的操作，并在块执行后取消通道。如果需要遍历通道而不使用通道，则应改用常规 for 循环。
    squares.consumeEach { println(it) }
    println("Done!")
}

//==========
// 扩展函数 开启单个协程
private fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++) // infinite stream of integers starting from 1
}

private fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
    for (x in numbers) send(x * x)
}

private fun channelProduceNumbers(time: Int) = runBlocking {
    val numbers = produceNumbers() // produces integers from 1 and on
    val squares = square(numbers) // squares integers
    repeat(time) {
        println(squares.receive()) // print first five
    }
    println("Done!") // we are done
    coroutineContext.cancelChildren() // cancel children coroutines
}

///=============
private fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
    for (x in numbers) {
        delay(200L)
//        println("numbers $x $prime")
        if (x % prime != 0) send(x)
    }
}

private fun channelProduceRecNumber() = runBlocking {
    var cur = produceNumbers()
    repeat(10) { it ->
        var prime = cur.receive() + 1
        println("${it}:$prime")
        cur = filter(cur, prime)
    }
    coroutineContext.cancelChildren() // cancel all children to let main finish
    println("end")
}

//========
private fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }
}

// send --> receive
private fun channelProRec() = runBlocking<Unit> {
    val producer = produceNumbers()
    // 开启五个协程
    repeat(5) { launchProcessor(it, producer) }
    delay(100)
    producer.cancel() // cancel producer coroutine and thus kill them all
}

//============


private fun main() {

//    channelNoBuffer()
//    channelBuffer(2)
//    produceSquares(2)
//    channelProduceNumbers(9)
//    channelProduceRecNumber()
//    channelProRec()
}